package edu.qdu.statics

import com.mongodb.casbah.{MongoClient, MongoClientURI}
import com.mongodb.casbah.commons.MongoDBObject
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

import java.text.SimpleDateFormat
import java.util.Date

/**
 * 基于评分统计
 * ● 统计所有历史数据当中每个商品的评分数次数，并存储到MongoDB的RateMoreProducts集合中。
 * ● 统计以月为单位拟每个商品的评分数次数，并存储到MongoDB的RateMoreRecentlyProducts集合中。
 * ● 统计每个商品的平均评分的分数，并存储到MongoDB的AverageProducts集合中。
 */
//封装mongoDB的连接信息
case class MongoConfig(url:String,db:String)
//封装用户商品评分数据
case class Rating(userID:Int,productID:Int,rating:Double,timestamp:Long)

/**
 * 静态推荐
 */
object StaticsRecommender {
  def main(args: Array[String]): Unit = {
    //Mongo的连接配置
    val configMap = Map(
      "url" -> "mongodb://spark01:27017/recommender",
      "db" -> "recommender"
    )

    //定义集合常量
    val RATING_COLLECTION = "Rating"
    val RATE_MORE_COLLECTION = "RateMoreProducts"
    val RATE_MORE_RECENTLY_COLLECTION = "RateMoreRecentlyProducts"
    val AVERAGE_COLLECTION = "AverageProducts"

    //定义一个隐士参数
    implicit val config = MongoConfig(configMap.getOrElse("url", ""), configMap.getOrElse("db", ""))

    //获取SparkSQL的上下文
    val spark = SparkSession.builder()
      .appName("dataloader")
      .master("local[*]") //如果往服务器部署，需要修改一下即可
      .getOrCreate()

    //引入SparkSQL的隐士转换
    import spark.implicits._
    //加载Rating数据集
    val ratingDF = spark
      .read
      .option("uri", config.url)
      .option("collection", RATING_COLLECTION)
      .format("com.mongodb.spark.sql")
      .load() //返回DataFrame
      .as[Rating] //将df转换成DS，
      .toDF()

    //将ratingDF注册成表
    ratingDF.createOrReplaceTempView("ratings")

    /**
     * 1、统计所有历史数据当中每个商品的评分数次数，并存储到MongoDB的RateMoreProducts集合中。
     */
    val rateMoreDF = spark.sql(
      """
        |select
        |productID productId,
        |count(*) count
        |from ratings
        |group by productID
        |""".stripMargin)
    rateMoreDF.show()

    //将历史商品中每个商品的评分次数统计存储到MongoDB中
    writeDataToMongoDB(rateMoreDF,RATE_MORE_COLLECTION)

    /**
     * 2、统计以月为单位拟每个商品的评分数次数，并存储到MongoDB的RateMoreRecentlyProducts集合中。
     */
    //自定义一个函数，将时间戳转换为YYYYMM
    val sdf = new SimpleDateFormat("YYYYMM")
    spark.udf.register("getMonth",(timeStamp:Long)=>sdf.format(new Date(timeStamp * 1000)))
    /*
    //计算每月每商品的评分次数---方法一
    val rateMoreRecentlyDF = spark.sql(
      """
        |select
        |productID productId,
        |count(*) count,
        |getMonth(timestamp) yearmonth
        |from ratings
        |group by productID,getMonth(timestamp)
        |""".stripMargin)*/
    //计算每月每商品的评分次数---方法二
    val rateMoreRecentlyDF = spark.sql(
      """
        |select
        |productID productId,
        |count(*) count,
        |from_unixtime(cast(timestamp as bigint),"yyyyMM") yearmonth
        |from ratings
        |group by productID,from_unixtime(cast(timestamp as bigint),"yyyyMM")
        |""".stripMargin)
    rateMoreRecentlyDF.show()

    //将最近的年月商品评分次数存储到MongoDB中
    writeDataToMongoDB(rateMoreRecentlyDF,RATE_MORE_RECENTLY_COLLECTION)

    /**
     * 3、统计每个商品的平均评分的分数，并存储到MongoDB的AverageProducts集合中。
     */
    val avgDF = spark.sql(
      """
        |select
        |productID productId,
        |avg(rating) avg
        |from ratings
        |group by productID
        |""".stripMargin)
    avgDF.show()

    //将avgDF存储到Mongo中
    writeDataToMongoDB(avgDF,AVERAGE_COLLECTION)

    //关闭spark
    spark.stop()
  }

  /**
   * 将productDF和ratingDF写入数库的集合中
   * @param spark
   * @param productDF
   * @param ratingDF
   * @param config
   */
  def writeDataToMongoDB(df: DataFrame,col:String)(implicit config:MongoConfig): Unit ={
    //获取Mongo的client
    val client = MongoClient(MongoClientURI(config.url))
    //获取操作集合
    val collection = client(config.db)(col)

    //删除集合
    collection.dropCollection()

    //将productDF写入collection中
    df
      .write
      .option("uri",config.url)
      .option("collection",col)
      .mode(SaveMode.Overwrite)
      .format("com.mongodb.spark.sql")
      .save()

    //对数据创建索引
    collection.createIndex(MongoDBObject("productID"->1))
  }
}